package com.lwq.imserver.code;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.auth0.jwt.interfaces.Claim;
import com.lwq.imserver.constants.ChannelAttr;
import com.lwq.imserver.model.TransferModel;
import com.lwq.imserver.service.MessageService;
import com.lwq.imserver.model.HeartBeats;
import com.lwq.imserver.model.ReplyBody;
import com.lwq.imserver.model.SendBody;
import com.lwq.imserver.util.RedisUtil;
import com.lwq.imserver.util.TokenUtil;
import com.lwq.imserver.util.WebSocketUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * Description:
 * Author: LiuWenQing
 * Datetime: 2022/1/17 17:00
 */
@Component
@Slf4j
public class EventChannelHandler extends SimpleChannelInboundHandler<Object> {

    private static final int PONG_TIME_OUT_COUNT = 3;

    private MessageService messageService;

    private RedisUtil redis;

    @Value("${netty.server.port}")
    private Integer HOST;

    public EventChannelHandler(MessageService messageService, RedisUtil redis) {
        this.messageService = messageService;
        this.redis = redis;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object o) throws InterruptedException {
        if (o instanceof HttpRequest) {
            doHandlerHttpRequest(ctx, (HttpRequest) o);
        } else if (o instanceof WebSocketFrame) {
            doHandlerWebSocketFrame(ctx, (WebSocketFrame) o);
        }
    }

    /**
     * websocket消息处理
     */
    private void doHandlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame msg) throws InterruptedException {
        if (msg instanceof TextWebSocketFrame) {
            TextWebSocketFrame textWebSocketFrame = (TextWebSocketFrame) msg;
            if (ctx.channel().isWritable()) {
                String text = textWebSocketFrame.text();
                ReplyBody replyBody = null;
                try {
                    //JSON解析
                    replyBody = JSONObject.parseObject(text, ReplyBody.class);
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("消息体解析出错");
                    ctx.close();
                    return;
                }
                if(StringUtils.isEmpty(replyBody.getType())){
                    ctx.close();
                    return ;
                }
                switch (replyBody.getType()) {
                    // 心跳
                    case ReplyBody.HEARTBEATS: {
                        ctx.channel().writeAndFlush(new TextWebSocketFrame(
                                JSON.toJSONString(new TextWebSocketFrame(JSON.toJSONString(new SendBody(HeartBeats.PONG, SendBody.HEARTBEATS))))));
                        break;
                    }
                    // 撤回
                    case ReplyBody.RECALL: {
                        messageService.del(replyBody);
                        break;
                    }
                    // 发送信息
                    case ReplyBody.SEND: {
                        messageService.send(replyBody);
                    }
                }
            }
        }
    }


    /**
     * 连接上服务器
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        System.out.println(WebSocketUtil.getUserName(ctx.channel()) + " ======= >" + ctx.channel().id() + "已连接上服务器");
    }

    /**
     * 通道活跃
     *
     * @param ctx
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ctx.channel().attr(ChannelAttr.UID).set(ctx.channel().id().asShortText());
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    /**
     * 不活跃的通道
     *
     * @param ctx
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        System.out.println(ctx.channel().id() + "已失去链接");
    }

    /**
     * 断开连接
     *
     * @param ctx
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        log.info(WebSocketUtil.getUserName(ctx.channel()) + "----------" + ctx.channel().id() + "已断开连接=========");
        String userName = WebSocketUtil.getUserName(ctx.channel());
        if(StringUtils.isNotEmpty(userName)){
            WebSocketUtil.broadCast(new SendBody("",TransferModel.LOGOUT,WebSocketUtil.getUserName(ctx.channel())));
        }
        WebSocketUtil.removeChannelByChannel(ctx.channel());
        release(ctx);
    }

    /**
     * 连接异常需要关闭相关资源
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error(cause.getMessage(), cause);
        WebSocketUtil.removeChannelByChannel(ctx.channel());
        release(ctx);
    }


    /**
     * wetsocket第一次连接握手
     */
    private void doHandlerHttpRequest(ChannelHandlerContext ctx, HttpRequest msg) {
        // http 解码失败
        if (!msg.getDecoderResult().isSuccess()) {
            sendHttpResponse(ctx, (FullHttpRequest) msg, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
        String token = WebSocketUtil.getToken(msg);
        if (StringUtils.isEmpty(token)) {
            ctx.writeAndFlush(new TextWebSocketFrame("缺少用户token,请登录"));
            release(ctx);
            return;
        }

        Map<String, Claim> map;
        try {
            map = TokenUtil.verifyToken(token);
        } catch (Exception e) {
            e.printStackTrace();
            sendHttpResponse(ctx, (FullHttpRequest) msg, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            release(ctx);
            return;
        }
        String userId = map.get("userId").asString();
        if(redis.get(userId) == null || redis.get(userId).isEmpty()){
            if(WebSocketUtil.getChannelMap().containsKey(userId)){
                WebSocketUtil.getChannelMap().remove(userId);
            }
            sendHttpResponse(ctx, (FullHttpRequest) msg, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            release(ctx);
            return;
        }

        if (WebSocketUtil.getChannelMap().containsKey(userId)) {
//            sendHttpResponse(ctx, (FullHttpRequest) msg, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
//            release(ctx);
//            return;
            WebSocketUtil.removeChannelByKey(userId);
        }

        Channel channel = ctx.channel();
        //TODO 能获取用户信息说明是登录用户 继续其他业务
        WebSocketUtil.getChannelMap().put(userId, channel);
        WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory("ws://" + msg.headers().get("Host") + HOST, null, false,
                65536 * 100);
        WebSocketServerHandshaker handshaker = factory.newHandshaker(msg);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
            return;
        }
        //进行连接
        handshaker.handshake(channel, (FullHttpRequest) msg);
    }

    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
        // 返回应答给客户端
        if (res.getStatus().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
        }
        // 如果是非Keep-Alive，关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }


    /**
     * netty内置检测心跳类
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (!(evt instanceof IdleStateEvent)) {
            return;
        }
        IdleStateEvent idleEvent = (IdleStateEvent) evt;
        String uid = ctx.channel().attr(ChannelAttr.UID).get();
        /*
         * 关闭未认证的连接
         */
        if (idleEvent.state() == IdleState.WRITER_IDLE && uid == null) {
            ctx.close();
            return;
        }

        /*
         * 已经认证的连接发送心跳请求
         */
        if (idleEvent.state() == IdleState.WRITER_IDLE && uid != null) {

            Integer pingCount = ctx.channel().attr(ChannelAttr.PING_COUNT).get();
            ctx.channel().attr(ChannelAttr.PING_COUNT).set(pingCount == null ? 1 : pingCount + 1);
            SendBody sendBody = new SendBody();
            sendBody.setType(SendBody.HEARTBEATS);
            sendBody.setContent(HeartBeats.PONG);
            ctx.channel().writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(sendBody)));
            return;
        }

        /*
         * 如果心跳请求发出30秒内没收到响应，则关闭连接
         */
        Integer pingCount = ctx.channel().attr(ChannelAttr.PING_COUNT).get();
        if (idleEvent.state() == IdleState.READER_IDLE && pingCount != null && pingCount >= PONG_TIME_OUT_COUNT) {
            ctx.close();
            log.info("{} pong timeout." , ctx.channel());
        }
    }

    private void release(ChannelHandlerContext ctx) {
        if (ctx != null) {
            ctx.close();
            ctx.channel().close();
        }
    }


}
